package com.wyg.ep_timer.task;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.wyg.ep_timer.entity.Article;
import com.wyg.ep_timer.entity.CompanyNum;
import com.wyg.ep_timer.entity.DataResult;
import com.wyg.ep_timer.feign.HttpUtil;
import com.wyg.ep_timer.mapper.*;
import com.wyg.ep_timer.phoenix.Query;
import com.wyg.ep_timer.service.ArticleEntityService;
import com.wyg.ep_timer.service.ArticleKwService;
import com.wyg.epcommonbase.common.util.BeanMapper;
import com.wyg.epcommonbase.common.util.StringUtil;
import com.wyg.epkafkastarter.config.KafkaConfigProperties;
import com.wyg.epkafkastarter.constant.Topic;
import com.wyg.epkafkastarter.consumer.impl.BatchConsumerImpl;
import io.netty.util.concurrent.RejectedExecutionHandler;
import io.netty.util.concurrent.RejectedExecutionHandlers;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.json.JSONException;
import org.json.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;

/**
 * Description:
 *
 * @author WangYongGan
 * @since 2022/6/1 14:04
 */
@Slf4j
@Configuration
@EnableScheduling
public class DealHbaseTask {
    @Autowired
    Query query;

    public DealHbaseTask(KafkaConfigProperties kafkaConfigProperties) {
        start();
    }

    public void start() {
        BatchConsumerImpl.start(Topic.EpArticle, records -> {
            List<Integer> ids = records.stream().map(record -> Integer.parseInt(record.value())).collect(Collectors.toList());
            final List<Article> articles = query.articleFromPhoenix(ids);
            dealData(articles);
        });

    }

    @Autowired
    DataResultMapper dataResultMapper;
    @Autowired
    CompanyNumMapper companyNumMapper;
    @Autowired
    ArticleKwService articleKwService;
    @Autowired
    ArticleEntityService articleEntityService;
    @Autowired
    ArticleEntityMapper articleEntityMapper;
    @Autowired
    ArticleKwMapper articleKwMapper;


    private void dealData(List<Article> dataSet) {
        int n = dataSet.size();
        if (n > 0) {
            log.error("处理" + n + "条数据");
        } else {
            log.error("没有数据，暂停10分钟");
            return;
        }
        for (Article data : dataSet) {
            //文本长度小于10 不合理
            log.info("处理id:" + data.getId() + "\ttext:" + data.getRawText().substring(0, 20));
            //标记已经处理
            final List<String> entities = deal(data);
            synchronized (Resource.lock) {
                for (String entity : entities) {
                    Integer num = (Integer) Resource.cache.get(entity);
                    num = num == null ? 0 : num;
                    Resource.cache.set(entity, num + 1);
                }
            }

        }
    }

    private List<String> deal(Article data) {
        //关键字
//        articleKwService.saveArticleKeyWord(data);
        String title = data.getTitle().equals("None") ? "" : data.getTitle() + ",";
        data.setTitle(title);
        List<String> regEntities = new ArrayList<>();
        List<String> entities = new ArrayList<>();
        JSONObject jsonObject = new JSONObject();
        DataResult dataResult = BeanMapper.copyByClass(data, DataResult.class);
        try {
            jsonObject.put("text", data.getTitle() + data.getRawText());
            //获得entity
            entities = HttpUtil.sendPostGetEntities(jsonObject);
            //对没给entity分析情感 得到reg_entities
            for (String i : entities) {
                jsonObject.put("entity", i);
                if (HttpUtil.setPostEmotionOfEntity(jsonObject)) {
                    dataResult.setIsReg(1);
                    regEntities.add(jsonObject.getString("entity"));
                }
            }
        } catch (JSONException | IOException e) {
            e.printStackTrace();
        }
        //更新Article 插入result

        dataResult.setEntity(StringUtil.trimList(entities));
        dataResult.setRegEntity(StringUtil.trimList(regEntities));
        //更新entityAll
        entities.forEach(entity -> {
            updateEntity(entity, false);
        });
        regEntities.forEach(entity -> {
            updateEntity(entity, true);
        });
        dataResult.setArticleId(data.getId());
        dataResult.setId(null);
        dataResultMapper.insert(dataResult);
        return entities;
    }

    /**
     * @Description:
     * @Params: mapper, entity, 实例
     * @return:
     **/
    @SneakyThrows
    private void updateEntity(String entity, boolean reg) {
        CompanyNum company = companyNumMapper.selectOne(new QueryWrapper<CompanyNum>().eq("name", entity));
        if (company == null) {
            companyNumMapper.insert(new CompanyNum(entity, 1));
        } else {
            if (reg)
                company.setRegNum(company.getRegNum() + 1);
            else company.setAllNum(company.getAllNum() + 1);
            companyNumMapper.update(company, new QueryWrapper<CompanyNum>().eq("name", entity));
        }
    }

}
